package org.springboot3.project.demoapi.impl;

import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.config.annotation.DubboService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springboot3.project.demoapi.StreamDemoService;

@DubboService(interfaceClass = StreamDemoService.class, version = "1.0")
public class StreamDemoServiceImpl implements StreamDemoService {

    private static final Logger log = LoggerFactory.getLogger(StreamDemoServiceImpl.class);

    /**
     * 服务端流
     * @param requestId requestId
     * @param response 响应流
     */
    @Override
    public void serverStream(String requestId, StreamObserver<String> response) {
        for (int i = 0; i < 5; i++) {
            response.onNext(requestId + ": message" + i);
        }
        response.onCompleted();
    }

    /**
     * 客户端流
     * @param response
     * @return
     */
    @Override
    public StreamObserver<String> clientStream(StreamObserver<String> response) {
        return new StreamObserver<String>() {
            @Override
            public void onNext(String data) {
                log.info("Received from client: " + data);
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("Error occurred: " + throwable.getMessage());
                throwable.printStackTrace();
            }

            @Override
            public void onCompleted() {
                log.info("Received from client: Completed");
                response.onNext("message received successfully");
                response.onCompleted();
            }
        };
    }

    /**
     * 双向流
     * @param response
     * @return
     */
    @Override
    public StreamObserver<String> bidirectionalStream(StreamObserver<String> response) {
        return new StreamObserver<String>() {
            @Override
            public void onNext(String data) {
                System.out.println("Received from client: " + data);
                response.onNext("Echo: " + data);
            }

            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void onCompleted() {
                response.onCompleted();
            }
        };
    }
}
